#include <unistd.h>
#include <stddef.h>
#include <pthread.h>
#include <glib.h>
#include "ocsplit.h"
#include "ocmapmerge.h"
#include "octhreadpool.h"
/*
struct _oc_map_merge_t
{
    oc_task_range_t         *input_task_range;
    gulong                  nthreads;
    oc_thread_pool_t        *tpool;
    GHashTable              **key_sum_ht;
    //oc_map_usr_func_t       map_usr_func;
    //oc_reduce_usr_func_t    reduce_usr_func;
    //oc_rkey_hash_func_t     rkey_hash_func;
    //oc_rkey_cmp_func_t      rkey_cmp_func;
};
*/
//static oc_task_range_t          *input_task_range = NULL;
static gulong                   nthreads = 0;
static oc_thread_pool_t         *tpool = NULL;
static GHashTable               **key_sum_ht = NULL;
static pthread_mutex_t          *key_sum_ht_mutex = NULL;
static oc_map_usr_func_t        map_usr_func = NULL;
static oc_reduce_usr_func_t     reduce_usr_func = NULL;
static oc_rkey_hash_func_t      rkey_hash_func = NULL;
static oc_rkey_cmp_func_t       rkey_cmp_func = NULL;
static oc_rkey_new_func_t       rkey_new_func = NULL;
static oc_rvalue_new_func_t     rvalue_new_func = NULL;
static oc_rkey_destroy_func_t   rkey_destroy_func = NULL;
static oc_rvalue_destroy_func_t rvalue_destroy_func = NULL;
static gboolean oc_rkey_cmp_bool_func ( gconstpointer rk0,
                                        gconstpointer rk1 );
static void oc_map_fireup ();
static void oc_merge_fireup ();
static void oc_map_thread_func ( gpointer task );
static void oc_merge_thread_func ( gpointer task );

void oc_map_merge_init (    //oc_task_range_t             *input_task_range,
                            gulong                      nts,
                            oc_map_usr_func_t           musr_func,
                            oc_reduce_usr_func_t        rusr_func,
                            oc_rkey_hash_func_t         rkhash_func,
                            oc_rkey_cmp_func_t          rkcmp_func,
                            oc_rkey_new_func_t          rknew_func,
                            oc_rvalue_new_func_t        rvnew_func,
                            oc_rkey_destroy_func_t      rkdestroy_func,
                            oc_rvalue_destroy_func_t    rvdestroy_func  )
                            //gboolean                    merge_switch    )
{
    nthreads = sysconf ( _SC_NPROCESSORS_ONLN );
    if ( nts > 0 && nts < nthreads )
    {
        nthreads = nts;
    }

    map_usr_func = musr_func;
    reduce_usr_func = rusr_func;
    rkey_hash_func = rkhash_func;
    rkey_cmp_func = rkcmp_func;
    rkey_new_func = rknew_func;
    rvalue_new_func = rvnew_func;
    rkey_destroy_func = rkdestroy_func;
    rvalue_destroy_func = rvdestroy_func;

    key_sum_ht = ( GHashTable ** ) g_slice_alloc ( sizeof ( GHashTable * ) * nthreads );
    key_sum_ht_mutex = ( pthread_mutex_t * ) g_slice_alloc ( sizeof ( pthread_mutex_t ) * nthreads );

    gulong i;
    for ( i = 0; i < nthreads; ++i )
    {
        key_sum_ht[i] = g_hash_table_new_full ( rkey_hash_func,
                                                oc_rkey_cmp_bool_func,
                                                rkey_destroy_func,
                                                rvalue_destroy_func );
        pthread_mutex_init ( & key_sum_ht_mutex[i], NULL );
    }

    tpool = oc_thread_pool_new (    nthreads );
                                    //oc_map_thread_func,
                                    //input_task_range );
}


void oc_map_merge_fireup(   //oc_map_merge_t *map_merge,
                            oc_task_range_t             *input_task_range,
                            //gulong                      nts,
                            gboolean                    merge_switch
                        )
{
    oc_thread_pool_set_task_range ( tpool, input_task_range );
    oc_thread_pool_set_usr_func ( tpool, oc_map_thread_func );
    oc_map_fireup ();

    if ( merge_switch )
    {
        oc_thread_pool_set_task_range ( tpool, input_task_range );
        oc_thread_pool_set_usr_func ( tpool, oc_merge_thread_func );
        oc_merge_fireup ();
    }
    else
    {
    }

}

static void oc_map_fireup ()
{
    oc_thread_pool_fireup (  tpool );
    oc_thread_pool_barrier (  tpool );
}

static void oc_merge_fireup ()
{
}

static
gboolean oc_rkey_cmp_bool_func (    gconstpointer rk0,
                                    gconstpointer rk1 )
{
    return rkey_cmp_func ( rk0, rk1) == 0 ? TRUE : FALSE;
}

static
void oc_map_thread_func ( gpointer task )
{

    oc_split_t *split = ( oc_split_t * )task;
    gulong i;
    for ( i = 0; i < oc_split_get_granularity ( split ); ++i )
    {
        oc_key_val_pair_t *kvpair = oc_split_get_kvpair( split, i );
        gpointer key = oc_key_val_pair_get_key ( kvpair );
        gpointer value = oc_key_val_pair_get_value ( kvpair );
        map_usr_func ( key, value );
    }
}

static
void oc_merge_thread_func ( gpointer task )
{
}

void oc_emit_intermediate ( gpointer rkey,
                            gpointer rvalue )
{
    gulong i = oc_thread_get_queue_index();
    gpointer rvalue_in_ht;
    pthread_mutex_lock ( & key_sum_ht_mutex[i] );
    if ( ( rvalue_in_ht = g_hash_table_lookup ( key_sum_ht[i], rkey ) ) )
    {
        pthread_mutex_unlock ( & key_sum_ht_mutex[i] );
        reduce_usr_func ( rvalue, rvalue_in_ht );
    }
    else
    {
        g_hash_table_insert ( key_sum_ht[i], rkey, rvalue );
        pthread_mutex_unlock ( & key_sum_ht_mutex[i] );
    }
}

void oc_map_merge_get_result()
{
}

void oc_map_merge_destroy()
{
    gulong i;
    for ( i = 0; i < nthreads; ++i )
    {
        g_hash_table_destroy ( key_sum_ht[i] );
        pthread_mutex_destroy ( & key_sum_ht_mutex[i] );
    }
    g_slice_free1 ( sizeof ( GHashTable * ) * nthreads, key_sum_ht );
}

